Skip to content

3.1 Agents

本节介绍 LangChain 中的 Agent(代理)概念和使用。


什么是 Agent?

Agent(代理) 是 LangChain 的核心概念。它将语言模型与工具结合,创建能够:

  • 推理任务
  • 决定使用哪些工具
  • 迭代地解决问题

"Agents combine language models with tools to create systems that can reason about tasks, decide which tools to use, and iteratively work towards solutions."


创建 Agent

使用 create_agent 函数创建生产就绪的 Agent:

python
from langchain.agents import create_agent
from langchain_core.tools import tool

@tool
def search(query: str) -> str:
    """搜索信息"""
    return f"搜索结果: {query}"

@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    return str(eval(expression))

agent = create_agent(
    "gpt-4o",
    tools=[search, calculate],
    system_prompt="你是一个有帮助的助手"
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "100 * 25 等于多少?"}]}
)

核心组件

1. 模型(Model)

Agent 的"大脑",负责推理和决策。

静态模型:创建时固定

python
agent = create_agent("gpt-4o", tools=tools)

动态模型:运行时选择

python
from langchain.agents import model_middleware

@model_middleware
def select_model(state, config):
    """根据任务复杂度选择模型"""
    if is_complex_task(state):
        return "gpt-4o"
    return "gpt-3.5-turbo"

2. 工具(Tools)

扩展 Agent 能力的可调用函数:

python
@tool
def get_current_time() -> str:
    """获取当前时间"""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件"""
    # 实现发送逻辑
    return f"邮件已发送给 {to}"

Agent 支持:

  • 单个请求中多次工具调用
  • 并行执行工具
  • 基于前一次结果动态选择工具
  • 自定义错误处理和重试

3. 系统提示词(System Prompt)

定义 Agent 的行为和个性:

python
system_prompt = """你是一个专业的数据分析师。

你的职责:
1. 分析用户提供的数据
2. 生成清晰的报告
3. 提供可行的建议

注意事项:
- 使用专业但易懂的语言
- 数据结论要有依据
"""

agent = create_agent(
    "gpt-4o",
    tools=[analyze_data, generate_chart],
    system_prompt=system_prompt
)

动态提示词

python
from langchain.agents import dynamic_prompt

@dynamic_prompt
def context_aware_prompt(state, config):
    """根据上下文生成提示词"""
    user_role = state.get("user_role", "general")

    if user_role == "admin":
        return "你是管理员助手,可以执行高级操作..."
    return "你是普通用户助手..."

ReAct 模式

Agent 遵循 ReAct(Reasoning + Acting) 模式:

用户: 北京和上海哪个城市更热?

Agent 思考: 我需要获取两个城市的温度来比较
Agent 行动: 调用 get_weather("北京")
观察结果: 北京 25度

Agent 思考: 已获取北京温度,现在需要上海的
Agent 行动: 调用 get_weather("上海")
观察结果: 上海 30度

Agent 思考: 两个温度都有了,可以比较
Agent 回答: 上海更热,30度比北京的25度高5度。

高级功能

结构化输出

让 Agent 返回固定格式的数据:

python
from dataclasses import dataclass
from langchain.agents import ToolStrategy

@dataclass
class AnalysisResult:
    summary: str
    confidence: float
    recommendations: list[str]

agent = create_agent(
    "gpt-4o",
    tools=[analyze],
    structured_output=ToolStrategy(AnalysisResult)
)

result = agent.invoke({"messages": [...]})
analysis = result["structured_response"]  # AnalysisResult 对象

自定义状态

扩展 AgentState 存储额外信息:

python
from langchain.agents import AgentState

class MyAgentState(AgentState):
    user_id: str
    session_data: dict
    analysis_count: int = 0

agent = create_agent(
    "gpt-4o",
    tools=[my_tools],
    state_schema=MyAgentState
)

中间件

在模型调用前后处理:

python
from langchain.agents import before_model, after_model

@before_model
def log_input(state, config):
    """记录输入"""
    print(f"输入: {state['messages'][-1].content}")
    return state

@after_model
def log_output(state, response, config):
    """记录输出"""
    print(f"输出: {response.content}")
    return state

agent = create_agent(
    "gpt-4o",
    tools=[my_tools],
    middleware=[log_input, log_output]
)

流式输出

实时显示 Agent 执行过程:

python
for event in agent.stream(
    {"messages": [{"role": "user", "content": "分析这份数据"}]},
    stream_mode="updates"
):
    if "llm" in event:
        print("思考中:", event["llm"])
    elif "tools" in event:
        print("执行工具:", event["tools"])

调用方式

invoke - 同步调用

python
result = agent.invoke(
    {"messages": [{"role": "user", "content": "你好"}]}
)
print(result["messages"][-1].content)

stream - 流式调用

python
for chunk in agent.stream({"messages": [...]}):
    print(chunk)

batch - 批量调用

python
results = agent.batch([
    {"messages": [{"role": "user", "content": "问题1"}]},
    {"messages": [{"role": "user", "content": "问题2"}]},
])

停止条件

Agent 在以下情况停止:

  1. 模型生成最终回复(不再调用工具)
  2. 达到最大迭代次数
  3. 遇到停止信号
python
agent = create_agent(
    "gpt-4o",
    tools=[my_tools],
    max_iterations=10,  # 最多迭代 10 次
)

最佳实践

实践说明
明确的系统提示词清晰定义 Agent 的角色和边界
工具描述准确让模型理解何时使用每个工具
错误处理工具中实现适当的错误处理
限制迭代次数防止无限循环
使用日志便于调试和监控

上一节3.0 Core Components

下一节3.2 Models

基于 MIT 许可证发布。内容版权归作者所有。